package com.passion.interceptor;

import com.alibaba.fastjson.JSONObject;
import com.passion.utils.DateTimeUtil;
import org.apache.flume.Context;
import org.apache.flume.Event;
import org.apache.flume.interceptor.Interceptor;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;

/**
 * @program flume-component
 * @Description 向header添加时间戳
 * @Author limeihua
 * @Date 2024/09/02 15:59
 * @Version 1.0
 */
public class TimeStampInterceptor implements Interceptor {
    public String chooseTSPosition;

    public TimeStampInterceptor(String chooseTSPosition) {
        this.chooseTSPosition = chooseTSPosition;
    }

    // 初始化操作
    @Override
    public void initialize() {

    }

    // 拦截
    @Override
    public Event intercept(Event event) {
        byte[] body = event.getBody();
        Map<String, String> headers = event.getHeaders();
        String bodyString = new String(body);
        String ts;
        try {
            JSONObject jsonObject = JSONObject.parseObject(bodyString);
            String[] split = chooseTSPosition.split("\\.");
            for (int i = 0; i < split.length - 1; i++) {
                jsonObject = jsonObject.getJSONObject(split[i]);
            }
            ts = jsonObject.getString(split[split.length - 1]);
            try {
                Long.parseLong(ts);
            }catch (Exception e){
                ts = DateTimeUtil.toTimestamp(ts.substring(0, 19));
            }
        } catch (Exception e) {
            return null;
        }
        headers.put("timestamp", ts);
        event.setHeaders(headers);
        return event;
    }

    @Override
    public List<Event> intercept(List<Event> list) {
        List<Event> events = new ArrayList<>();
        for (Event event : list) {
            Event intercept = intercept(event);
            if (intercept != null) {
                events.add(intercept);
            }
        }
        return events;
    }

    @Override
    public void close() {

    }


    public static class Builder implements Interceptor.Builder {

        public String chooseTSPosition;

        @Override
        public Interceptor build() {
            return new TimeStampInterceptor(chooseTSPosition);
        }

        // 配置参数
        @Override
        public void configure(Context context) {
            // 获取添加拦截器字段名，字段顺序
            chooseTSPosition = context.getString("chooseTSPosition");
        }
    }
}
